package com.bawei.flink.adsstatic;

import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.kudu.client.*;

import java.util.ArrayList;

public class KuduSinkFunction extends RichSinkFunction<Tuple5<String, String,String, Integer, Long>> {

    KuduClient client;
    KuduTable table;
    KuduSession kuduSession;
    Integer i = 0;

    @Override
    public void open(Configuration parameters) throws Exception {
        client = new KuduClient.KuduClientBuilder("cdh1:7051").defaultSocketReadTimeoutMs(10000).build();
        table = client.openTable("ads_log");
        kuduSession = client.newSession();
        kuduSession.setFlushMode(SessionConfiguration.FlushMode.MANUAL_FLUSH);
        kuduSession.setMutationBufferSpace(1000);
    }



    @Override
    public void invoke(Tuple5<String, String, String, Integer, Long> value, Context context) throws Exception {
        Insert insert = table.newInsert();
        insert.getRow().addString("startTime",value.f0);
        insert.getRow().addString("endTime",value.f1);
        insert.getRow().addString("city",value.f2);
        insert.getRow().addInt("adsId",value.f3);
        insert.getRow().addLong("cnts",value.f4);
        i ++;
        kuduSession.apply(insert);
        if(i >= 5) {
            kuduSession.flush();
            i = 0;
        }
    }

    @Override
    public void close() throws Exception {
        kuduSession.flush();
        kuduSession.close();
        client.close();
    }
}
